package com.kalvan.mq.consumer;

import cn.hutool.core.lang.Assert;
import com.alibaba.fastjson.JSON;
import com.kalvan.mq.MqCondition;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Mq消费者管理器
 *
 * @author chenliang
 */
@Conditional({MqCondition.class})
@Configuration
@Slf4j
@Getter
@Setter
public class MqConsumerManager implements InitializingBean, ApplicationContextAware {
    /**
     * 主题配置
     */
    private Map<String, String> topics;

    /**
     * MQ地址
     */
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;


    /**
     * 消费者Map
     */
    private Map<String, DefaultMQPushConsumer> consumerMap = new HashMap<>();

    private ApplicationContext applicationContext;

    /**
     * 消息处理器缓存
     */
    private Map<String, Map<String, MsgHandler>> msgHandlerCache = new HashMap<>();

    /**
     * 根据消息处理器创建消息consumer
     *
     * @param msgHandlerMetaData 消费处理器原数据
     */
    private void createMqConsumer(List<MsgHandlerMetaData> msgHandlerMetaData) {
        if (StringUtils.isEmpty(namesrvAddr)) {
            log.info("mq地址未配置，不创建消费者");
            return;
        }
        Map<String, ConsumerConfig> consumerConfigMap = mergeTopic(msgHandlerMetaData);
        //创建消费者
        for (Map.Entry<String, ConsumerConfig> entry : consumerConfigMap.entrySet()) {
            ConsumerConfig consumerConfig = entry.getValue();
            log.info("创建MQ消费者：{}", JSON.toJSON(consumerConfig));
            String topic = consumerConfig.getTopic();
            String tag = consumerConfig.getTag();
            String groupName = String.format("TM_%s_GROUP", topic);
            try {
                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
                consumer.setNamesrvAddr(namesrvAddr);
                consumer.subscribe(topic, tag);
                consumer.setMessageModel(consumerConfig.getMessageModel());
                //消费者并发能力优化
                consumer.setConsumeThreadMin(consumerConfig.getThreadMin());
                consumer.setConsumeThreadMax(consumerConfig.getThreadMax());
                consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
                    if (CollectionUtils.isEmpty(msgList)) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    int index = 0;
                    try {
                        for (; index < msgList.size(); index++) {
                            MessageExt msg = msgList.get(index);
                            log.debug("MQ消息：{}", JSON.toJSON(msg));
                            String msgTopic = msg.getTopic();
                            String msgTag = msg.getTags();
                            String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                            Map<String, MsgHandler> msgHandlerMap = msgHandlerCache.get(msgTopic);
                            if (CollectionUtils.isEmpty(msgHandlerMap)) {
                                log.warn("主题[{}]不存在消息处理器", msgTopic);
                                continue;
                            }
                            MsgHandler msgHandler = msgHandlerMap.get("*");
                            if (msgHandler != null) {
                                log.warn("[{}-{}]存在“*”消息处理器", msgTopic, msgTag);
                                boolean success = msgHandler.handle(content);
                                if (!success) {
                                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                }
                            }
                            msgHandler = msgHandlerMap.get(msgTag);
                            if (msgHandler != null) {
                                boolean success = msgHandler.handle(content);
                                if (!success) {
                                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                }
                            } else {
                                log.warn("未找到消息处理器[{}-{}]", msgTopic, msgTag);
                            }
                        }
                    } catch (Throwable e) {
                        log.error("消费数据异常", e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    } finally {
                        if (index < msgList.size()) {
                            context.setAckIndex(index + 1);
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                });
                consumer.start();
                consumerMap.put(topic, consumer);
            } catch (MQClientException e) {
                log.error("consumer 创建异常", e);
            }
        }
    }

    /**
     * 合并主题
     *
     * @param msgHandlerMetaData msgHandlerMetaData
     * @return Map<String, ConsumerConfig>
     */
    private Map<String, ConsumerConfig> mergeTopic(List<MsgHandlerMetaData> msgHandlerMetaData) {
        Map<String, ConsumerConfig> consumerConfigMap = new HashMap<>(10);
        for (MsgHandlerMetaData item : msgHandlerMetaData) {
            String topic = item.tagEnum().topic;
            ConsumerConfig consumerConfig = consumerConfigMap.get(topic);
            if (consumerConfig == null) {
                consumerConfig = new ConsumerConfig();
            }
            String metaDataTag = item.tagEnum().tag;
            if (!StringUtils.isEmpty(metaDataTag)) {
                StringBuilder tagStringBuilder = new StringBuilder(consumerConfig.getTag() == null ? "" : consumerConfig.getTag());
                if (tagStringBuilder.length() > 0) {
                    tagStringBuilder.append("||");
                }
                tagStringBuilder.append(metaDataTag);
                consumerConfig.setTag(tagStringBuilder.toString());
            }
            consumerConfig.setMessageModel(item.consumerMessageModel());
            consumerConfig.setThreadMax(item.consumerThreadMax());
            consumerConfig.setThreadMin(item.consumerThreadMin());
            consumerConfig.setTopic(item.tagEnum().topic);
            consumerConfigMap.put(topic, consumerConfig);
        }
        return consumerConfigMap;
    }

    @Override
    public void afterPropertiesSet() {
        List<MsgHandlerMetaData> msgHandlerMetaData = this.loadMsgHandler();
        if (!CollectionUtils.isEmpty(msgHandlerMetaData)) {
            this.createMqConsumer(msgHandlerMetaData);
        }
    }

    /**
     * 加载消息处理器
     */
    private List<MsgHandlerMetaData> loadMsgHandler() {
        List<MsgHandlerMetaData> msgHandlerMetaDataList = new ArrayList<>();
        Assert.notNull(applicationContext, "applicationContext 为空，消息处理器初始化失败");
        Map<String, MsgHandler> handlerMap = this.applicationContext.getBeansOfType(MsgHandler.class);
        if (handlerMap.isEmpty()) {
            log.warn("未发现消息处理器");
            return msgHandlerMetaDataList;
        }
        for (Map.Entry<String, MsgHandler> entry : handlerMap.entrySet()) {
            MsgHandler msgHandler = entry.getValue();
            MsgHandlerMetaData annotation = msgHandler.getClass().getAnnotation(MsgHandlerMetaData.class);
            if (annotation == null) {
                continue;
            }
            msgHandlerMetaDataList.add(annotation);
            String topic = annotation.tagEnum().topic;
            String tag = annotation.tagEnum().tag;
            if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(tag)) {
                log.warn("消息处理器Topic、tag都不能为空[{}-{}]", topic, tag);
                continue;
            }
            Map<String, MsgHandler> msgHandlerMap = msgHandlerCache.get(topic);
            if (msgHandlerMap == null) {
                msgHandlerMap = new HashMap<>(20);
            }
            String[] tagList = tag.split("\\|");
            for (String item : tagList) {
                msgHandlerMap.put(item, msgHandler);
            }
            msgHandlerCache.put(topic, msgHandlerMap);
        }
        return msgHandlerMetaDataList;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    @PreDestroy
    public void stop() {
        if (!CollectionUtils.isEmpty(consumerMap)) {
            for (Map.Entry<String, DefaultMQPushConsumer> entry : consumerMap.entrySet()) {
                DefaultMQPushConsumer consumer = entry.getValue();
                consumer.shutdown();
                log.info("关闭MQ消费者：{}{", consumer.getInstanceName());
            }
        }
    }
}